﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using Cassandra;
using LumenWorks.Framework.IO.Csv;
using System.Diagnostics;
using System.Threading.Tasks;

namespace CassyLoader
{
    class Loader
    {
        public Cluster Cluster { get; private set; }
        public ISession Session { get; private set; }

        private PreparedStatement InsertRowPreparedStatement;

        static void Main(string[] args)
        {
            Loader loader = new Loader();
            

            loader.Start();
        }


        private void PrepareStatements()
        {
            
            InsertRowPreparedStatement = Session.Prepare
                                        (
                                            "INSERT INTO feed.trplmorisk " +
                                            "(risk_id, instrument_name, effective_date, lob, region, cluster, sector, portfolio, strategy, accounting_type, product_group, notional_ccy, notional) " +
                                            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);"
                                        );
        }

        private void Start()
        {
            this.Connect("127.0.0.1");
            this.PrepareStatements();

            Stopwatch sw = new Stopwatch();

            sw.Start();
            
            List<Task> tasks = new List<Task>();

            using (CsvReader csv = new CsvReader(new StreamReader(@"C:\Users\~~~\Desktop\data.csv"), false))
            {
                int fieldCount = csv.FieldCount;
                string[] headers = csv.GetFieldHeaders();               

                while (csv.ReadNextRecord())
                {
                    try
                    {

                        Int64 ts =  GetUnixTimestamp(csv[1]);
                        BoundStatement boundStatement = InsertRowPreparedStatement.Bind(Guid.NewGuid(),csv[0].ToString(),                                                                                
                                                                                GetUnixTimestamp(csv[1]), csv[2].ToString(), csv[3].ToString(),
                                                                                csv[4].ToString(), csv[5].ToString(), csv[6].ToString(), csv[7].ToString(),
                                                                                csv[8].ToString(), csv[9].ToString(), 
                                                                                float.Parse(csv[10]), float.Parse(csv[11]));
                        //Session.Execute(boundStatement);
                        var resultSetFuture = Session.ExecuteAsync(boundStatement);
                        tasks.Add(resultSetFuture);
                    }
                    catch (Exception e)
                    {
                        throw;
                    }


                }
            }

            Task.WaitAll(tasks.ToArray());

            sw.Stop();
            Console.WriteLine("Operation took (sec): " + (sw.ElapsedMilliseconds / 1000));

            Console.ReadKey(); // pause the console before exiting

            this.Close();
        }


        public void Connect(String node)
        {
            Cluster = Cluster.Builder()
                     .AddContactPoint(node)
                     .Build();

            Session = Cluster.Connect();

            Console.WriteLine("Connected to cluster: " + Cluster.Metadata.ClusterName.ToString());

            foreach (var host in Cluster.Metadata.AllHosts())
            {
                Console.WriteLine("Data Center: " + host.Datacenter + ", " +
                    "Host: " + host.Address + ", " +
                    "Rack: " + host.Rack);
            }
        }


        public void Close()
        {
            Cluster.Shutdown();
        }


        public Int64 GetUnixTimestamp(string date)
        {            
            TimeSpan span = (DateTime.Parse(date) - new DateTime(1970, 1, 1, 0, 0, 0, 0).ToLocalTime());
            return (Int64) span.TotalSeconds * 1000; //multiply to get msecs
        }

    }
}
